package ru.spbau.database.group3.index;

import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import ru.spbau.database.group3.engine.Column;
import ru.spbau.database.group3.engine.Condition;
import ru.spbau.database.group3.engine.Relation;
import ru.spbau.database.group3.filesAndAccess.HeapFile;
import ru.spbau.database.group3.filesAndAccess.Page;
import ru.spbau.database.group3.filesAndAccess.Record;
import ru.spbau.database.group3.indexes.btree.BPlusTree;
import ru.spbau.database.group3.memoryManager.SpaceManager;
import ru.spbau.database.group3.parser.ConditionOperator;

public class BTreeIndex extends Index implements Serializable {
	
	private BPlusTree<IndexBTreeData, Set<Record>> indexTree;
	private List<Column> relationColumns;
	
	private List<Page> freePages = new LinkedList<Page>();
	private List<Page> fullPages = new LinkedList<Page>();

	public static final int DEFAULT_BUCKET_SIZE = 10;
	private static int bucketSize = DEFAULT_BUCKET_SIZE;
	private final int recordSize;
	private final String indexName;

	private Page allocNewPage() throws IOException {
		return SpaceManager.allocPage(indexName + Relation.DATA_EXT, recordSize);
	}
	
	private ByteBuffer constructKeyData(Record r) throws IOException {
        ByteBuffer entryData = ByteBuffer.allocate(recordSize);
        Column[] columns = indexKeyType.getColumns();
        for (Column c : columns) {
        	for (Column c1: relationColumns) {
        		if (c1.equals(c)) {
        			c1.getType().writeValueToByteBuffer(entryData, r.getColumnValue(c1));
        		}
        	}
        }
        return entryData;
	}
	
	private void deleteKey(Record key) throws IOException {
		int pid = key.getPid();
		for (Page p : freePages) {
			if (p.getId() == pid) {
				p.clearSlot(key);
				return;
			}
		}
		for (Page p : fullPages) {
			if (p.getId() == pid) {
				p.clearSlot(key);
				fullPages.remove(p);
				freePages.add(p);
				return;
			}			
		}
	}
	
	private Record insertKey(Record key) throws IOException{
		Page currentPage = null;
		if (freePages.size() == 0) {
			//no free pages
			currentPage = allocNewPage();
			freePages.add(currentPage);
		} else {
			currentPage = freePages.get(0);
		}
		ByteBuffer recordData = constructKeyData(key);
		Record result = currentPage.insertRecord(recordData);
		if (currentPage.isFull() && freePages.remove(currentPage)) {
			fullPages.add(currentPage);
		}
		return result;
	}
	
	public static int getBucketSize() {
		return bucketSize;
	}

	public static void setBucketSize(int arg) {
		bucketSize = arg;
	}

	public BTreeIndex(Set<Column> columns, boolean isUnique, String indexName, int recordSize, List<Column> relationColumns) throws IOException {
		super(columns, isUnique, relationColumns);
		this.indexName = indexName;
		this.recordSize = recordSize;
		this.relationColumns = relationColumns;
		indexTree = new BPlusTree<IndexBTreeData, Set<Record>>(bucketSize);
		Page firstPage = SpaceManager.allocPage(indexName + Relation.DATA_EXT, recordSize);
		freePages.add(firstPage);		
	}

	@Override
	public Collection<Record> getRecords(Condition c) {
		return getRecordsInSet(c);
	}

	public Set<Record> getRecordsInSet(Condition c) {
		ConditionOperator cop = c.getOperator();
		Set<Record> result = new HashSet<Record>();
		IndexBTreeData indexKeyData = new IndexBTreeData(new RecordSearchSample(c),indexKeyType);
		switch (cop) {
		case E:
			Set<Record> bufRes = indexTree.search(indexKeyData); 
			if (bufRes != null) {
				result.addAll(bufRes);
			}
			return result;
		case GEQ: {
			Collection<Set<Record>> resCollection = indexTree
					.searchMoreOrEqual(indexKeyData);
			for (Set<Record> pagesSet : resCollection) {
				if (pagesSet != null) {
					result.addAll(pagesSet);
				}
			}
			return result;
		}
		case GT: {
			Collection<Set<Record>> resCollection = indexTree
					.searchMore(indexKeyData);
			for (Set<Record> pagesSet : resCollection) {
				if (pagesSet != null) {
					result.addAll(pagesSet);
				}
			}
			return result;
		}
		case LEQ: {
			Collection<Set<Record>> resCollection = indexTree
					.searchLessOrEqual(indexKeyData);
			for (Set<Record> pagesSet : resCollection) {
				if (pagesSet != null) {
					result.addAll(pagesSet);
				}
			}
			return result;
		}
		case LT: {
			Collection<Set<Record>> resCollection = indexTree
					.searchLess(indexKeyData);
			for (Set<Record> pagesSet : resCollection) {
				if (pagesSet != null) {
					result.addAll(pagesSet);
				}
			}
			return result;
		}
		default:
			throw new UnsupportedOperationException(
					"Unsupported BTree index condition type!");
		}
	}

	@Override
	public void insertRecord(Record r) throws IOException {
		IndexBTreeData newKeyData = new IndexBTreeData(r, indexKeyType);
		Set<Record> pagesSet = indexTree.search(newKeyData);
		if (pagesSet != null) {
			if (IS_UNIQUE
					&& (pagesSet.size() > 1 || (pagesSet.size() > 0 && !pagesSet.contains(r)))) {
				throw new RuntimeException(
						"Inserted an entry that contradicts with index uniqueness.");
			}
			pagesSet.add(r);
		} else {
			pagesSet = new HashSet<Record>();
			//nothing was stored for such key => such key was not present in the index .d file
			//insert it
//			for (Column c : relationColumns) {
//				System.out.println(r.getColumnValue(c));
//			}
			IndexBTreeData keyData = new IndexBTreeData(insertKey(r), indexKeyType);
			pagesSet.add(r);
			indexTree.insert(keyData, pagesSet);
		}
	}

	@Override
	public Record deleteRecord(Record r) throws IOException {
		IndexBTreeData deleteKeyData = new IndexBTreeData(r, indexKeyType);
		Set<Record> pagesSet = indexTree.search(deleteKeyData);
		if (pagesSet != null && pagesSet.remove(r)) {
			//TODO: remove the record from data page
			if (pagesSet.isEmpty()) {
				//there are no entries for such key; delete the key
				IndexBTreeData key = indexTree.delete(deleteKeyData);
				deleteKey(key.getRecord());
			}
			return r;	
		}
		return null;
	}

	@Override
	public Collection<Record> getRecords(Collection<Condition> c) {
		Set<Record> resultSet = new HashSet<Record>();
		for (Condition condition : c) {
			Set<Record> conditionSet = getRecordsInSet(condition);
			if (resultSet.isEmpty()) {
				resultSet.addAll(conditionSet);
			} else {
				resultSet.retainAll(conditionSet);
			}
		}
		return resultSet;
	}

	public String getName() {
		return indexName;
	}
	
	@Override
	public String toString() {
		return indexTree.toString();
	}
}
